package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;


/**
 * 1.获取ods中日志数topic_log
 * 2.处理过滤topic_log
 * 3.对新老用户进行处理
 *      is_new:1
 *              状态=null 我要把当前人ts转换成年月日 更新到状态值
 *              状态！=null 判断当前数据的ts 和状态日期是否相同 如果不同更新数据
 *      is_new:0
 *              状态=null 将ts——1day 更新到状态中
 *              如果有值 不需要处理
 * 4.把主流数据拆分成五个流 分别存入五个主题中
 *      1.创建流式环境
 *      2.设置并行度
 *      3.运行jar 向topic——log主题发送数据
 *      4.从kafka读取主流数据
 *      5.数据清洗 转换结构
 *      6.将脏数据写出到kafka指定主题
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //todo 1.初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //todo 并行度
        env.setParallelism(1);
        //todo 2 启用状态后端
        //todo 3 从kafka读取主流数据
        String topic = "topic_log";
        String groupId = "base_log_consumer";

        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        source.print();

        //todo 4.数据清洗 转换结构
        OutputTag<String> dirtyStream = new OutputTag<String>("dirtyStream") {

        };

        env.execute();
    }
}
